package com.atguigu.dga.governance.service.impl;

import com.atguigu.dga.dolphinscheduler.bean.TDsTaskDefinition;
import com.atguigu.dga.dolphinscheduler.bean.TDsTaskInstance;
import com.atguigu.dga.dolphinscheduler.service.TDsTaskDefinitionService;
import com.atguigu.dga.dolphinscheduler.service.TDsTaskInstanceService;
import com.atguigu.dga.governance.assess.Assessor;
import com.atguigu.dga.governance.assess.impl.spec.TecOwnerAssessor;
import com.atguigu.dga.governance.bean.AssessParam;
import com.atguigu.dga.governance.bean.GovernanceAssessDetail;
import com.atguigu.dga.governance.bean.GovernanceMetric;
import com.atguigu.dga.governance.mapper.GovernanceAssessDetailMapper;
import com.atguigu.dga.governance.service.*;
import com.atguigu.dga.governance.util.SpringBeanProvider;
import com.atguigu.dga.meta.bean.TableMetaInfo;
import com.atguigu.dga.meta.service.TableMetaInfoService;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import java.util.stream.Collectors;

/**
 * <p>
 * 治理考评结果明细 服务实现类
 * </p>
 *
 * @author zhangjunyi
 * @since 2023-06-06
 */
@Service
@DS("dga")
public class GovernanceAssessDetailServiceImpl extends ServiceImpl<GovernanceAssessDetailMapper, GovernanceAssessDetail> implements GovernanceAssessDetailService {

    @Autowired
    TableMetaInfoService tableMetaInfoService;

    @Autowired
    GovernanceMetricService governanceMetricService;

    @Autowired
    TDsTaskInstanceService tDsTaskInstanceService;

    @Autowired
    TDsTaskDefinitionService tDsTaskDefinitionService;

    @Autowired
    GovernanceAssessTableService governanceAssessTableService;

    @Autowired
    GovernanceAssessTecOwnerService governanceAssessTecOwnerService;

    @Autowired
    GovernanceAssessGlobalService governanceAssessGlobalService;

    ThreadPoolExecutor threadPoolExecutor=new ThreadPoolExecutor(100,100,60, TimeUnit.SECONDS,new LinkedBlockingDeque<>());

    @Override
    public void assessAllTableMetrics(String assessDate) {
        //0.重复考评的话先将当天的考评结果清理掉
        remove(new QueryWrapper<GovernanceAssessDetail>().eq("assess_date", assessDate));

        //1.先获取所有最新的元数据TableMetaInfo包含（table_meta_info_extra`）  这个查询出来的是一个集合 集合中放的是需要考评的每一张表  因此需要遍历
        List<TableMetaInfo> tableMetaInfoAllList = tableMetaInfoService.getTableMetaInfoAllList();

        //获取到当天所有DS中任务运行的实例
        List<TDsTaskInstance> tDsTaskInstanceByDtList = tDsTaskInstanceService.getTDsTaskInstanceByDt(assessDate);

        //获取任务定义表的Map集合
        //获取到所有考评表的全表名的集合
        List<String> allAssessFullTableNameList = tableMetaInfoAllList.stream().map(tableMetaInfo -> tableMetaInfo.getSchemaName() + "." + tableMetaInfo.getTableName()).collect(Collectors.toList());
        Map<String, TDsTaskDefinition> tableDefinitionMapByDt = tDsTaskDefinitionService.getTableDefinitionMapByDt(allAssessFullTableNameList, assessDate);


        //将所有表的元数据封装为 K-V结构 k：库名+表名  v：这个表对应的TableMetaInfo数据
        HashMap<String, TableMetaInfo> tableMetaInfoHashMap = new HashMap<>();
        for (TableMetaInfo tableMetaInfo : tableMetaInfoAllList) {
            tableMetaInfoHashMap.put(tableMetaInfo.getSchemaName() + "." + tableMetaInfo.getTableName(), tableMetaInfo);
        }

        //2.先获取到考评指标governance_metric 在获取数据的时候要获取的是is_disabled 为 1 的指标  这个查询出来的是一个集合 集合中放的是需要考评的指标
        List<GovernanceMetric> governanceMetricList = governanceMetricService.list(new QueryWrapper<GovernanceMetric>().eq("is_disabled", "0"));

        //创建List集合用来存放考评明细
        ArrayList<GovernanceAssessDetail> governanceAssessDetails = new ArrayList<>(tableMetaInfoAllList.size()*governanceMetricList.size());

        //收集期货的集合（异步编排要用的集合）
        ArrayList<CompletableFuture<GovernanceAssessDetail>> completableFutures = new ArrayList<>();

        //3.进行双层for循环对每张表每个指标进行考评
        for (TableMetaInfo tableMetaInfo : tableMetaInfoAllList) {
            for (GovernanceMetric governanceMetric : governanceMetricList) {

                //获取具体某个考评指标的编码
                String metricCode = governanceMetric.getMetricCode();

                //创建考评参数对象
                AssessParam assessParam = new AssessParam();
                assessParam.setAssessDate(assessDate);
                assessParam.setGovernanceMetric(governanceMetric);
                assessParam.setTableMetaInfo(tableMetaInfo);
                assessParam.setTableMetaInfoMap(tableMetaInfoHashMap);
                assessParam.setTDsTaskInstanceListWithAssessDate(tDsTaskInstanceByDtList);
                assessParam.setTDsTaskDefinition(tableDefinitionMapByDt.get(tableMetaInfo.getSchemaName()+"."+tableMetaInfo.getTableName()));

                //通过模板模式再利用spring组件提取工具 获取到子对象
                Assessor assessor = SpringBeanProvider.getBean(metricCode, Assessor.class);
                //调用抽象类中主考评方法
                GovernanceAssessDetail governanceAssessDetail = assessor.mainAssess(assessParam);

                governanceAssessDetails.add(governanceAssessDetail);

            /*    CompletableFuture<GovernanceAssessDetail> completableFuture = CompletableFuture.supplyAsync(new Supplier<GovernanceAssessDetail>() {
                    @Override
                    public GovernanceAssessDetail get() {
                        return assessor.mainAssess(assessParam);
                    }
                },threadPoolExecutor);
                completableFutures.add(completableFuture);*/
                //如果按照这种方式写的话虽然能够实现需求但是不方便后期的维护，比如要新增一个考评指标那么就得修改代码
               /* if (governanceMetric.getMetricCode()=="TEC_OWNER"){
                    //对遍历出来的每个TableMetaInfo做考评
                }else if (governanceMetric.getMetricCode()=="TEC_OWNER"){

                }else if (){

                }*/
            }
        }

  /*      //期货的交付（异步编排） 将计算后的结果汇总起来
        List<GovernanceAssessDetail> governanceAssessDetailList = completableFutures.stream().map(governanceAssessDetailCompletableFuture -> governanceAssessDetailCompletableFuture.join()).collect(Collectors.toList());*/

        //4.将考评的明细保存到mysql表当中
//        saveOrUpdateBatch(governanceAssessDetailList);
        saveOrUpdateBatch(governanceAssessDetails);
    }

    @Override
    public void mainAssess(String assessDate) {
        long startTime = System.currentTimeMillis();
        //针对每个指标项做考评
        assessAllTableMetrics(assessDate);
        System.out.println("---------------计算时间："+(System.currentTimeMillis()-startTime));

        governanceAssessTableService.calcGovernanceAssessTable(assessDate);
        governanceAssessTecOwnerService.calcGovernanceAssess(assessDate);
        governanceAssessGlobalService.calcGovernanceAssess(assessDate);
    }


}
